package com.zhang.sparkstreaming_2

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Duration, StreamingContext}

import java.util.UUID

/**
 * @title:
 * @author: zhang
 * @date: 2022/2/21 11:12 
 */
object SparkStreaming03_Queue {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark streaming")
    val ssc = new StreamingContext(conf, Duration(3000L))

    //todo 自定义数据源读取数据
    ssc.receiverStream(new MyReceiver())
      .print()



    //启动采集器
    ssc.start()
    ssc.awaitTermination()
  }

  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {
    private var running = true

    override def onStart(): Unit = {
      while (running) {
        val uuid = UUID.randomUUID().toString
        store(uuid)
        Thread.sleep(1000L)
      }
    }

    override def onStop(): Unit = {
      running = false
    }
  }
}
